%% copy riak-0.14.2 source to learn dynamo implementation and erlang
%% author: lichuang
%% copy source from riak_object.erl
%% @doc container for edynamo data and metadata
-module(edynamo_object).
-include("../include/riak_kv_wm_raw.hrl").

-type key() :: binary().
-type bucket() :: binary().
-type value() :: term().

-record(r_content, {metadata :: dict(),
                    value    :: term()}).

%% Opaque container for edynamo objects, a.k.a. edynamo_object()
-record(r_object, {
	  bucket :: bucket(),
	  key	 :: key(),
	  contents :: [#r_content{}],
	  vclock :: vclock:vclock(),
	  updatemetadata=dict:store(clean, true, dict:new()) :: dict(),
	  updatevalue :: term()}).
-type edynamo_object() :: #r_object{}.
-define(MAX_KEY_SIZE, 65536).
-export([new/3, new/4]).
-export([increment_vclock/2, increment_vclock/3]).
-export([update_metadata/2, update_value/2, bucket/1,
         reconcile/2, ancestors/1, key/1,
         apply_updates/1, get_metadatas/1, get_update_metadata/1]).
-export([get_contents/1]).
-export([set_vclock/2, vclock/1]).

%% @doc Constructor for new edynamo objects.
-spec new(Bucket::bucket(), Key::key(), Value::value()) -> edynamo_object().
new(B, K, V) when is_binary(B), is_binary(K) ->
  new(B, K, V, no_initial_metadata).

%% @doc Constructor for new edynamo objects with an initial content-type.
-spec new(Bucket::bucket(), Key::key(), Value::value(), string() | dict()) ->edynamo_object().
new(B, K, V, C) when is_binary(B), is_binary(K), is_list(C) ->
  new(B, K, V, dict:from_list([{?MD_CTYPE, C}]));

%% @doc Constructor for new edynamo objects with an initial metadata dict.
new(B, K, V, MD) when is_binary(B), is_binary(K) ->
  case size(K) > ?MAX_KEY_SIZE of
    true ->
      throw({error, key_too_large});
    false ->
      case MD of
	no_initial_metadata ->
	  Contents = [#r_content{metadata = dict:new(), value = V}],
	  #r_object{bucket=B,key=K,
		    contents=Contents,vclock=vclock:fresh()};
	_ ->
	  Contents = [#r_content{metadata=MD, value=V}],
	  #r_object{bucket=B, key=K, updatemetadata=MD,
		    contents=Contents,vclock=vclock:fresh()}
	end
  end.

%% @doc  Increment the entry for ClientId in O's vclock.
-spec increment_vclock(edynamo_object(), vclock:vclock_node()) ->edynamo_object().
increment_vclock(Object=#r_object{}, ClientId) ->
  Object#r_object{vclock=vclock:increment(ClientId, Object#r_object.vclock)}.

%% @doc  Increment the entry for ClientId in O's vclock.
-spec increment_vclock(edynamo_object(), vclock:vclock_node(), vclock:timestamp()) -> edynamo_object().
increment_vclock(Object=#r_object{}, ClientId, Timestamp) ->
  Object#r_object{vclock=vclock:increment(ClientId, Timestamp, Object#r_object.vclock)}.

%% @spec update_metadata(edynamo_object(), dict()) -> 
update_metadata(Object=#r_object{}, M) ->
  Object#r_object{updatemetadata=dict:erase(clean, M)}.

%% @spec update_value(edynamo_object(), value()) -> edynamo_object()
%% @doc  Set the updated value of an object to V
update_value(Object=#r_object{}, V) ->
  Object#r_object{updatevalue=V}.

%% @spec set_vclock(edynamo_object(), vclock:vclock()) -> edynamo_object()
%% @doc  INTERNAL USE ONLY.  Set the vclock of edynamo_object O to V.
set_vclock(Object=#r_object{}, VClock) ->
  Object#r_object{vclock=VClock}.

%% @spec get_contents(edynamo_object()) -> [{dict(), value()}]
%% @doc  Return the contents (a list of {metadata, value} tuples) for
%%       this edynamo_object.
get_contents(#r_object{contents=Contents}) ->
  [{Content#r_content.metadata, Content#r_content.value} ||
      Content <- Contents].

%% @spec bucket(edynamo_object()) -> bucket()
%% @doc Return the containing bucket for this edynamo_object.
bucket(#r_object{bucket=Bucket}) ->
  Bucket.

%% @spec get_update_metadata(edynamo_object()) -> dict()
%% @doc  Return the updated metadata of this edynamo_object.
get_update_metadata(#r_object{updatemetadata = UM}) ->
  UM.

%% @spec get_metadatas(edynamo_object()) -> [dict()]
%% @doc  Return a list of the metadata values for this edynamo_object.
get_metadatas(#r_object{contents=Contents}) ->
  [Content#r_content.metadata || Content <- Contents].

%% @spec apply_updates(edynamo_object()) -> edynamo_object()
%% @doc  Promote pending updates (made with the update_value() and
%%       update_metadata() calls) to this edynamo_object.
apply_updates(Object = #r_object{}) ->
  VL = case Object#r_object.updatevalue of
        undefined ->
          [C#r_content.value || C <- Object#r_object.contents];
        _ ->
          [Object#r_object.updatevalue]
       end,
  MD = case dict:find(clean, Object#r_object.updatemetadata) of
        {ok, _} ->
          MDs = [C#r_content.metadata || C <- Object#r_object.contents],
          case Object#r_object.updatevalue of
            undefined -> MDs;
            _ -> [hd(MDs)]
          end;
        error ->
          [dict:erase(clean, Object#r_object.updatemetadata) || _X <- VL]
       end,
  Contents = [#r_content{metadata=M, value = V} || {M, V} <- lists:zip(MD, VL)],
  Object#r_object{contents = Contents,
                  updatemetadata = dict:store(clean, true, dict:new()),
                  updatevalue    = undefined}.

%% @spec vclock(edynamo_object()) -> vclock:vclock()
%% @doc  Return the vector clock for this edynamo_object.
vclock(#r_object{vclock=VClock}) ->
  VClock.

%% @spec ancestors([edynamo_object()]) -> [edynamo_object()]
%% @doc  Given a list of edynamo_object()s, return the objects that are pure
%%       ancestors of other objects in the list, if any.  The changes in the
%%       objects returned by this function are guaranteed to be reflected in
%%       the other objects in Objects, and can safely be discarded from the list
%%       without losing data.
ancestors(Objects) ->
  ToRemove = [[O2 || O2 <- Objects,
      vclock:descends(O1#r_object.vclock, O2#r_object.vclock),
      (vclock:descends(O2#r_object.vclock, O1#r_object.vclock) == false)]
    || O1 <- Objects],
  lists:flatten(ToRemove).    

%% @spec reconcile([edynamo_object()], boolean()) -> edynamo_object()
%% @doc  Reconcile a list of riak objects.  If AllowMultiple is true,
%%       the edynamo_object returned may contain multiple values if Objects
%%       contains sibling versions (objects that could not be syntactically
%%       merged).   If AllowMultiple is false, the riak_object returned will
%%       contain the value of the most-recently-updated object, as per the
%%       X-Riak-Last-Modified header.
reconcile(Objects, AllowMultiple) ->
  RObjs = reconcile(Objects),
  AllContents = lists:flatten([O#r_object.contents || O <- RObjs]),
  Contents = case AllowMultiple of
    false ->
      [most_recent_content(AllContents)];
    true ->
      AllContents
  end,
  VClock = vclock:merge([O#r_object.vclock || O <- RObjs]),
  HdObj = hd(RObjs),
  HdObj#r_object{contents=Contents,vclock=VClock,
                 updatemetadata=dict:store(clean, true, dict:new()),
                 updatevalue=undefined}.
%% @spec reconcile([edynamo_object()]) -> [edynamo_object()]
reconcile(Objects) ->
  All = sets:from_list(Objects),
  Del = sets:from_list(ancestors(Objects)),
  remove_duplicate_objects(sets:to_list(sets:subtract(All, Del))).

remove_duplicate_objects(Os) -> rem_dup_objs(Os, []).
rem_dup_objs([], Acc) -> Acc;
rem_dup_objs([O | Rest], Acc) ->
  EqO = [AO || AO <- Acc, edynamo_object:equal(AO, O) =:= true],
  case EqO of
    [] -> rem_dup_objs(Rest, [O | Acc]);
    _  -> rem_dup_objs(Rest, Acc)
  end.

most_recent_content(AllContents) ->
  hd(lists:sort(fun compare_content_dates/2, AllContents)).

compare_content_dates(C1, C2) ->
  D1 = dict:fetch(<<"X-Riak-Last-Modified">>, C1#r_content.metadata),
  D2 = dict:fetch(<<"X-Riak-Last-Modified">>, C2#r_content.metadata),
  %% true if C1 was modifed later than C2
  Cmp1 = edynamo_util:compare_dates(D1, D2),
  %% true if C2 was modifed later than C1
  Cmp2 = edynamo_util:compare_dates(D2, D1),
  %% check for deleted objects
  Del1 = dict:is_key(<<"X-Riak-Deleted">>, C1#r_content.metadata),
  Del2 = dict:is_key(<<"X-Riak-Deleted">>, C2#r_content.metadata),

  SameDate = (Cmp1 =:= Cmp2),
  case {SameDate, Del1, Del2} of
    {false, _, _} ->
      Cmp1;
    {true, true, false} ->
      false;
    {true, false, true} ->
      true;
    _ ->
      %% Dates equal and either both present or both deleted, compare
      %% by opaque contents.
      C1 < C2
  end.

%% @spec key(edynamo_object()) -> key()
%% @doc  Return the key for this edynamo_object.
key(#r_object{key=Key}) -> Key.
